package com.shelpe.services.sinbad.service;

import java.util.ArrayList;
import java.util.List;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import rx.Observable;
import rx.Subscriber;

@Slf4j
public class SubscribleScheduler<O> {

	private Handler<O> handler;

	private List<Observable<O>> readys;
	private List<Observable<O>> finishes;

	private int concurrency = 3;

	public void addObservable(Observable<O> ob) {
		if (this.readys == null) {
			this.readys = new ArrayList<Observable<O>>();
		}
		this.readys.add(ob);
	}

	public void setupHandler(final Handler<O> handler) {
		this.handler = handler;
	}

	public void setupConcurrency(int con) {
		this.concurrency = con;
	}

	public void invoke(final Callback callback) {
		log.debug("total {} observables in {} concurrency", this.readys.size(), this.concurrency);
		final Observable<O> last = Observable.merge(this.readys, this.concurrency);

		Subscriber<O> subscriber = new Subscriber<O>() {

			@Override
			public void onCompleted() {
				long start = System.currentTimeMillis();
				callback.calll(null);
				log.info("invoke onCompleted cost {} ms", System.currentTimeMillis() - start);
			}

			@Override
			public void onError(Throwable e) {
				long start = System.currentTimeMillis();
				SchedulerInterruptExcepion ex = new SchedulerInterruptExcepion();
				ex.total = readys.size();
				if (finishes != null) {
					ex.complete = finishes.size();
				}
				callback.calll(ex);
				log.info("invoke onError cost {} ms", System.currentTimeMillis() - start);
			}

			@Override
			public void onNext(O t) {
				long start = System.currentTimeMillis();
				handler.handle(t);
				this.invokeNext();
				log.info("invoke onNext cost {} ms", System.currentTimeMillis() - start);
			}

			private void invokeNext() {
				move2Finish();
			}

		};

		last.subscribe(subscriber);
	}

	private synchronized boolean move2Finish() {
		Observable<O> next = null;
		if (this.readys != null && this.readys.size() > 0) {
			next = this.readys.iterator().next();
		}

		if (next != null) {
			log.debug("move to finish");
			if (this.finishes == null) {
				this.finishes = new ArrayList<Observable<O>>();
			}
			this.finishes.add(next);
		}
		log.debug("remain {} observables", this.readys.size());

		return this.readys.size() == this.finishes.size();
	}

	public static interface Handler<O> {
		void handle(O t);
	}

	public static interface Callback {
		void calll(SchedulerInterruptExcepion ex);
	}

	@Data
	public static class SchedulerInterruptExcepion extends RuntimeException {
		private int total;
		private int complete;
	}

}
